package cn.sunchis.demo.kafka;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;

import cn.sunchis.pvuv.sdk.config.PvuvServiceConfig;

import java.util.Random;

@Controller
@SpringBootApplication
public class KafkaDemoApplication {
	public static Logger logger = LoggerFactory.getLogger(KafkaDemoApplication.class);
	
	private Random random = new Random();
	
	@Autowired
    private KafkaTemplate<String, String> template;
	
	public static void main(String[] args) {
		SpringApplication.run(KafkaDemoApplication.class, args);
	}
	
	@ResponseBody
	@RequestMapping("/send")
	public String send(String topic, String key, String data) {
        template.send("topic", "number", String.valueOf(random.nextInt()));
        return "success";
    }
	
	@KafkaListener(id="topic", topics = "topic")
    public void listenT2(ConsumerRecord<?, ?> cr) throws Exception {
        logger.info("{} - {} : {}", cr.topic(), cr.key(), cr.value());
    }
	
	/**
	 * PVUV统计消费端
	 * @param cr
	 * @throws Exception
	 */
	@KafkaListener(id=PvuvServiceConfig.KAFKA_PVUV_TOPIC, topics=PvuvServiceConfig.KAFKA_PVUV_TOPIC)
    public void consumePvuv(ConsumerRecord<?, ?> cr) throws Exception {
        logger.info("{} - {} : {}", cr.topic(), cr.key(), cr.value());
    }
}
